package transaction

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/state/store"
)

// TransactionStatus represents the transaction status
type TransactionStatus int

const (
	// TransactionStatusPending represents a pending transaction
	TransactionStatusPending TransactionStatus = iota
	// TransactionStatusCommitted represents a committed transaction
	TransactionStatusCommitted
	// TransactionStatusAborted represents an aborted transaction
	TransactionStatusAborted
	// TransactionStatusTimedOut represents a timed out transaction
	TransactionStatusTimedOut
)

// String returns the string representation of the transaction status
func (s TransactionStatus) String() string {
	switch s {
	case TransactionStatusPending:
		return "pending"
	case TransactionStatusCommitted:
		return "committed"
	case TransactionStatusAborted:
		return "aborted"
	case TransactionStatusTimedOut:
		return "timed_out"
	default:
		return "unknown"
	}
}

// TransactionIsolationLevel represents the transaction isolation level
type TransactionIsolationLevel int

const (
	// TransactionIsolationLevelReadUncommitted represents the read uncommitted isolation level
	TransactionIsolationLevelReadUncommitted TransactionIsolationLevel = iota
	// TransactionIsolationLevelReadCommitted represents the read committed isolation level
	TransactionIsolationLevelReadCommitted
	// TransactionIsolationLevelRepeatableRead represents the repeatable read isolation level
	TransactionIsolationLevelRepeatableRead
	// TransactionIsolationLevelSerializable represents the serializable isolation level
	TransactionIsolationLevelSerializable
)

// String returns the string representation of the transaction isolation level
func (l TransactionIsolationLevel) String() string {
	switch l {
	case TransactionIsolationLevelReadUncommitted:
		return "read_uncommitted"
	case TransactionIsolationLevelReadCommitted:
		return "read_committed"
	case TransactionIsolationLevelRepeatableRead:
		return "repeatable_read"
	case TransactionIsolationLevelSerializable:
		return "serializable"
	default:
		return "unknown"
	}
}

// TransactionConfig represents the transaction configuration
type TransactionConfig struct {
	// Timeout is the transaction timeout
	Timeout time.Duration
	// IsolationLevel is the transaction isolation level
	IsolationLevel TransactionIsolationLevel
	// ReadOnly is the read-only flag
	ReadOnly bool
}

// DefaultTransactionConfig returns the default transaction configuration
func DefaultTransactionConfig() *TransactionConfig {
	return &TransactionConfig{
		Timeout:        time.Second * 30,
		IsolationLevel: TransactionIsolationLevelReadCommitted,
		ReadOnly:       false,
	}
}

// Operation represents a transaction operation
type Operation struct {
	// Type is the operation type
	Type string
	// State is the state name
	State string
	// Key is the key
	Key string
	// Value is the value
	Value []byte
}

// Transaction represents a transaction
type Transaction struct {
	// ID is the transaction ID
	ID string
	// Status is the transaction status
	Status TransactionStatus
	// Config is the transaction configuration
	Config *TransactionConfig
	// Operations is the operations
	Operations []*Operation
	// StartTime is the start time
	StartTime time.Time
	// EndTime is the end time
	EndTime time.Time
	// store is the state store
	store store.StateStore
	// snapshot is the state snapshot
	snapshot map[string]map[string][]byte
	// mu is the mutex
	mu sync.RWMutex
}

// NewTransaction creates a new transaction
func NewTransaction(id string, config *TransactionConfig, store store.StateStore) *Transaction {
	if config == nil {
		config = DefaultTransactionConfig()
	}

	return &Transaction{
		ID:        id,
		Status:    TransactionStatusPending,
		Config:    config,
		Operations: make([]*Operation, 0),
		StartTime: time.Now(),
		store:     store,
		snapshot:  make(map[string]map[string][]byte),
	}
}

// Get gets a value from the state
func (t *Transaction) Get(ctx context.Context, stateName string, key string) ([]byte, error) {
	t.mu.RLock()
	defer t.mu.RUnlock()

	// Check if the transaction is still active
	if t.Status != TransactionStatusPending {
		return nil, fmt.Errorf("transaction is not active: %s", t.Status)
	}

	// Check if the value is in the snapshot
	if stateSnapshot, ok := t.snapshot[stateName]; ok {
		if value, ok := stateSnapshot[key]; ok {
			return value, nil
		}
	}

	// Get the state
	state, err := t.store.GetState(ctx, stateName)
	if err != nil {
		return nil, fmt.Errorf("failed to get state: %w", err)
	}

	// Check the state type
	kvState, ok := state.(store.KVState)
	if !ok {
		return nil, errors.New("state is not a KV state")
	}

	// Get the value
	value, err := kvState.Get(ctx, key)
	if err != nil {
		return nil, fmt.Errorf("failed to get value: %w", err)
	}

	// Add the value to the snapshot
	if _, ok := t.snapshot[stateName]; !ok {
		t.snapshot[stateName] = make(map[string][]byte)
	}
	t.snapshot[stateName][key] = value

	return value, nil
}

// Set sets a value in the state
func (t *Transaction) Set(ctx context.Context, stateName string, key string, value []byte) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	// Check if the transaction is still active
	if t.Status != TransactionStatusPending {
		return fmt.Errorf("transaction is not active: %s", t.Status)
	}

	// Check if the transaction is read-only
	if t.Config.ReadOnly {
		return errors.New("transaction is read-only")
	}

	// Add the operation
	t.Operations = append(t.Operations, &Operation{
		Type:  "set",
		State: stateName,
		Key:   key,
		Value: value,
	})

	// Update the snapshot
	if _, ok := t.snapshot[stateName]; !ok {
		t.snapshot[stateName] = make(map[string][]byte)
	}
	t.snapshot[stateName][key] = value

	return nil
}

// Delete deletes a value from the state
func (t *Transaction) Delete(ctx context.Context, stateName string, key string) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	// Check if the transaction is still active
	if t.Status != TransactionStatusPending {
		return fmt.Errorf("transaction is not active: %s", t.Status)
	}

	// Check if the transaction is read-only
	if t.Config.ReadOnly {
		return errors.New("transaction is read-only")
	}

	// Add the operation
	t.Operations = append(t.Operations, &Operation{
		Type:  "delete",
		State: stateName,
		Key:   key,
	})

	// Update the snapshot
	if _, ok := t.snapshot[stateName]; !ok {
		t.snapshot[stateName] = make(map[string][]byte)
	}
	t.snapshot[stateName][key] = nil

	return nil
}

// Commit commits the transaction
func (t *Transaction) Commit(ctx context.Context) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	// Check if the transaction is still active
	if t.Status != TransactionStatusPending {
		return fmt.Errorf("transaction is not active: %s", t.Status)
	}

	// Apply all operations
	for _, op := range t.Operations {
		// Get the state
		state, err := t.store.GetState(ctx, op.State)
		if err != nil {
			// Abort the transaction
			t.Status = TransactionStatusAborted
			t.EndTime = time.Now()
			return fmt.Errorf("failed to get state: %w", err)
		}

		// Check the state type
		kvState, ok := state.(store.KVState)
		if !ok {
			// Abort the transaction
			t.Status = TransactionStatusAborted
			t.EndTime = time.Now()
			return errors.New("state is not a KV state")
		}

		// Apply the operation
		switch op.Type {
		case "set":
			err = kvState.Set(ctx, op.Key, op.Value)
			if err != nil {
				// Abort the transaction
				t.Status = TransactionStatusAborted
				t.EndTime = time.Now()
				return fmt.Errorf("failed to set value: %w", err)
			}
		case "delete":
			err = kvState.Delete(ctx, op.Key)
			if err != nil {
				// Abort the transaction
				t.Status = TransactionStatusAborted
				t.EndTime = time.Now()
				return fmt.Errorf("failed to delete value: %w", err)
			}
		default:
			// Abort the transaction
			t.Status = TransactionStatusAborted
			t.EndTime = time.Now()
			return fmt.Errorf("unknown operation type: %s", op.Type)
		}
	}

	// Update the transaction status
	t.Status = TransactionStatusCommitted
	t.EndTime = time.Now()

	return nil
}

// Abort aborts the transaction
func (t *Transaction) Abort(ctx context.Context) error {
	t.mu.Lock()
	defer t.mu.Unlock()

	// Check if the transaction is still active
	if t.Status != TransactionStatusPending {
		return fmt.Errorf("transaction is not active: %s", t.Status)
	}

	// Update the transaction status
	t.Status = TransactionStatusAborted
	t.EndTime = time.Now()

	return nil
}

// GetStatus returns the transaction status
func (t *Transaction) GetStatus() TransactionStatus {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.Status
}

// GetOperations returns the transaction operations
func (t *Transaction) GetOperations() []*Operation {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.Operations
}

// GetStartTime returns the transaction start time
func (t *Transaction) GetStartTime() time.Time {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.StartTime
}

// GetEndTime returns the transaction end time
func (t *Transaction) GetEndTime() time.Time {
	t.mu.RLock()
	defer t.mu.RUnlock()
	return t.EndTime
}

// TransactionManager represents the transaction manager
type TransactionManager struct {
	// transactions is the transactions
	transactions map[string]*Transaction
	// store is the state store
	store store.StateStore
	// mu is the mutex
	mu sync.RWMutex
}

// NewTransactionManager creates a new transaction manager
func NewTransactionManager(store store.StateStore) *TransactionManager {
	return &TransactionManager{
		transactions: make(map[string]*Transaction),
		store:        store,
	}
}

// Begin begins a new transaction
func (m *TransactionManager) Begin(ctx context.Context, id string, config *TransactionConfig) (*Transaction, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Check if the transaction already exists
	if _, ok := m.transactions[id]; ok {
		return nil, fmt.Errorf("transaction %s already exists", id)
	}

	// Create a new transaction
	tx := NewTransaction(id, config, m.store)
	m.transactions[id] = tx

	return tx, nil
}

// Get gets a transaction
func (m *TransactionManager) Get(id string) (*Transaction, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Get the transaction
	tx, ok := m.transactions[id]
	if !ok {
		return nil, fmt.Errorf("transaction %s not found", id)
	}

	return tx, nil
}

// Commit commits a transaction
func (m *TransactionManager) Commit(ctx context.Context, id string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Get the transaction
	tx, ok := m.transactions[id]
	if !ok {
		return fmt.Errorf("transaction %s not found", id)
	}

	// Commit the transaction
	err := tx.Commit(ctx)
	if err != nil {
		return fmt.Errorf("failed to commit transaction: %w", err)
	}

	return nil
}

// Abort aborts a transaction
func (m *TransactionManager) Abort(ctx context.Context, id string) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Get the transaction
	tx, ok := m.transactions[id]
	if !ok {
		return fmt.Errorf("transaction %s not found", id)
	}

	// Abort the transaction
	err := tx.Abort(ctx)
	if err != nil {
		return fmt.Errorf("failed to abort transaction: %w", err)
	}

	return nil
}

// CleanUp cleans up completed transactions
func (m *TransactionManager) CleanUp(ctx context.Context, olderThan time.Duration) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Get the current time
	now := time.Now()

	// Clean up completed transactions
	for id, tx := range m.transactions {
		// Check if the transaction is completed
		if tx.Status == TransactionStatusCommitted || tx.Status == TransactionStatusAborted || tx.Status == TransactionStatusTimedOut {
			// Check if the transaction is old enough
			if now.Sub(tx.EndTime) > olderThan {
				// Remove the transaction
				delete(m.transactions, id)
			}
		}
	}

	return nil
}

// CheckTimeouts checks for transaction timeouts
func (m *TransactionManager) CheckTimeouts(ctx context.Context) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Get the current time
	now := time.Now()

	// Check for transaction timeouts
	for _, tx := range m.transactions {
		// Check if the transaction is still active
		if tx.Status == TransactionStatusPending {
			// Check if the transaction has timed out
			if now.Sub(tx.StartTime) > tx.Config.Timeout {
				// Update the transaction status
				tx.Status = TransactionStatusTimedOut
				tx.EndTime = now
			}
		}
	}

	return nil
}

// GetTransactions returns all transactions
func (m *TransactionManager) GetTransactions() []*Transaction {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Get all transactions
	transactions := make([]*Transaction, 0, len(m.transactions))
	for _, tx := range m.transactions {
		transactions = append(transactions, tx)
	}

	return transactions
}

// GetActiveTransactions returns all active transactions
func (m *TransactionManager) GetActiveTransactions() []*Transaction {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Get active transactions
	transactions := make([]*Transaction, 0)
	for _, tx := range m.transactions {
		if tx.Status == TransactionStatusPending {
			transactions = append(transactions, tx)
		}
	}

	return transactions
}

// GetCompletedTransactions returns all completed transactions
func (m *TransactionManager) GetCompletedTransactions() []*Transaction {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Get completed transactions
	transactions := make([]*Transaction, 0)
	for _, tx := range m.transactions {
		if tx.Status != TransactionStatusPending {
			transactions = append(transactions, tx)
		}
	}

	return transactions
}
